package com.scheduler.servicediscover;

import com.alibaba.fastjson.JSON;
import com.scheduler.actor.ActorMessage;
import com.scheduler.actor.WorkerActor;
import com.scheduler.actor.WorkerActorEventBus;
import com.scheduler.utils.okhttp.OkHttpClient;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.net.*;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 服务发现心跳上报，可以知道有几台机器在线
 */
@Slf4j
public class ServiceDiscoverHeartbeat {

    public static final ServiceDiscoverHeartbeat INSTANCE = new ServiceDiscoverHeartbeat();

    public static final AtomicBoolean initialized = new AtomicBoolean(false);
    /**
     * 上一次心跳打印时间，用于控制打印间隔
     */
    private static final AtomicLong pre_log_time = new AtomicLong(0);
    private static final long log_interval = 60 * 1000;
    /**
     * 多网卡的情况下，可能有时候需要配置一下前缀，通常不需要
     */
    @Setter
    @Getter
    private String ipPrefix;

    @Setter
    @Getter
    private int port;

    @Setter
    @Getter
    private ServiceDiscoverRegistrationCenter serviceDiscoverRegistrationCenter;

    private final WorkerActorEventBus workerActorEventBus = WorkerActorEventBus.defaultWorkerActorEventBus;

    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private ServiceDiscoverHeartbeat() {
    }

    public String getIp() {
        return getLocalIpByNetCard();
    }

    /**
     * 获取活动的服务器列表
     */
    public List<Worker> getActiveServerList() {
        return serviceDiscoverRegistrationCenter.getAllWorkers();
    }

    public void start() {
        if (initialized.compareAndSet(false, true)) {
            log.info("初始化服务器心跳上报开始");
            serviceDiscoverRegistrationCenter.init();
            initServer();
            initPingWorkerActor();
            initDaemon();
            addShutdownHook();
            log.info("初始化服务器心跳上报完成");
        }
    }

    public void stop() {
        // 只需要把服务心跳停掉就好了，workerActorEventBus 就不会再发送请求过来。
        try {
            log.info("停机关闭心跳线程池：ServiceDiscoverHeartbeat");
            scheduledExecutorService.shutdown();
            boolean b = scheduledExecutorService.awaitTermination(30, TimeUnit.SECONDS);
            if (b) {
                log.info("停机关闭心跳线程池成功：ServiceDiscoverHeartbeat");
            } else {
                log.info("停机关闭心跳线程池失败：ServiceDiscoverHeartbeat");
            }
        } catch (InterruptedException e) {
            log.error("停机关闭心跳线程池异常", e);
        }
    }

    private void initServer() {
        workerActorEventBus.start(port);
    }

    private void initPingWorkerActor() {
        log.info("初始化WorkActor：name=PING，Dispatcher=PING");
        WorkerActor workerActor = new WorkerActor("PING")
                .addDispatcher("PING", actorMessage -> WorkerActorEventBus.OK);
        workerActorEventBus.registerCallbackWorkActor(workerActor);
    }

    private void initDaemon() {
        scheduledExecutorService.scheduleWithFixedDelay(this::doHeartbeat, 60, 3, TimeUnit.SECONDS);
    }

    private void doHeartbeat() {
        try {
            // 获取IP地址
            String ip = getIp();
            // ping 通才上报
            String ping = ping(ip);
            if (WorkerActorEventBus.OK.equals(ping)) {
                // 上报心跳
                serviceDiscoverRegistrationCenter.doHeartbeat(ip);
                long nowTime = System.currentTimeMillis();
                if (nowTime - pre_log_time.get() > log_interval) {
                    log.info("上报服务器IP地址成功, ip={}", ip);
                    pre_log_time.set(nowTime);
                }
            } else {
                log.info("上报服务器IP地址跳过，本机服务器未PING通, ip={}", ip);
            }
        } catch (Throwable t) {
            log.error("上报服务器IP地址异常", t);
        }
    }

    private String ping(String ip) {
        String url = String.format(workerActorEventBus.getRpcUrlPrefix(), ip, workerActorEventBus.getPort()) + workerActorEventBus.getRpcUrl();
        ActorMessage actorMessage = ActorMessage.builder()
                .auth(workerActorEventBus.getAuth())
                .taskType(ActorMessage.TaskType.MAP_SIMPLE.name())
                .actorName("PING")
                .actorDispatcher("PING")
                .object(null)
                .build();
        return OkHttpClient.doPost(url, JSON.toJSONString(actorMessage), OkHttpClient.Header.of("accept", WorkerActorEventBus.ACCEPT));
    }

    /**
     * 直接根据第一个网卡地址作为其内网ipv4地址，避免返回 127.0.0.1
     *
     * @return
     */
    private String getLocalIpByNetCard() {
        try {
            for (Enumeration<NetworkInterface> e = NetworkInterface.getNetworkInterfaces(); e.hasMoreElements(); ) {
                NetworkInterface item = e.nextElement();
                for (InterfaceAddress address : item.getInterfaceAddresses()) {
                    if (item.isLoopback() || !item.isUp()) {
                        continue;
                    }
                    if (address.getAddress() instanceof Inet4Address) {
                        Inet4Address inet4Address = (Inet4Address) address.getAddress();
                        String hostAddress = inet4Address.getHostAddress();
                        // 注意匹配方式为前缀匹配，所以不要加0，也不要加星号
                        if (StringUtils.isNotBlank(ipPrefix) && !hostAddress.startsWith(ipPrefix)) {
                            continue;
                        }
                        return hostAddress;
                    }
                }
            }
            return InetAddress.getLocalHost().getHostAddress();
        } catch (SocketException | UnknownHostException e) {
            throw new RuntimeException(e);
        }
    }

    private void addShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
    }
}
